package vip.meeet.jdk.concurent;


import org.apache.commons.lang3.StringUtils;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;


public class TaskHelper {

    private static SendDataTask sendDataTask;
    private static BlockingQueue<String> dataQueue = new LinkedBlockingQueue<>(100000);

    public static void main(String[] args) {
        sendDataTask = new SendDataTask();
        Thread thread = new Thread(sendDataTask, "SendDataTask");
        thread.setDaemon(true);
        thread.start();
    }

    public void submit(String data) {
        if (StringUtils.isNotBlank(data)) {
            dataQueue.add(data);
            sendDataTask.signalQueue();
        }
    }

    private static class SendDataTask implements Runnable {

        private ReentrantLock lock = new ReentrantLock();
        private Condition notEmpty = lock.newCondition();

        private int checkPeriod = 10 * 1000;
        private volatile boolean stop = false;

        @Override
        public void run() {
            while (!stop) {
                if (isEmpty()) {
                    awaitQueue();
                }
                try {
                    Thread.sleep(checkPeriod);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                if (stop) {
                    break;
                }
                check();
            }
        }

        public void check() {
            boolean timedOut = false;
            List<String> dataList = new ArrayList<>();
            for (; ; ) {
                if (timedOut) {
                    return;
                }
                try {
                    String data = dataQueue.poll(100, TimeUnit.MILLISECONDS);
                    if (StringUtils.isBlank(data)) {
                        timedOut = true;
                        if (dataList.size() > 0) {
                            sendData(dataList);
                        }
                    } else {
                        dataList.add(data);
                        if (dataList.size() > 100) {
                            sendData(dataList);
                        }
                    }
                } catch (InterruptedException ignore) {
                }
            }
        }

        public void stop() {
            this.stop = true;
            //执行关闭逻辑
        }

        private boolean isEmpty() {
            return dataQueue.size() == 0;
        }

        private void awaitQueue() {
            boolean flag = lock.tryLock();
            if (flag) {
                try {
                    notEmpty.await();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } catch (Exception e) {
                    e.printStackTrace();
                } finally {
                    lock.unlock();
                }
            }
        }

        public void signalQueue() {
            boolean flag = false;
            try {
                flag = lock.tryLock(100, TimeUnit.MILLISECONDS);
                if (flag)
                    notEmpty.signalAll();
            } catch (InterruptedException e) {
                e.printStackTrace();
            } catch (Exception e) {
                e.printStackTrace();
            } finally {
                if (flag)
                    lock.unlock();
            }
        }

        public void sendData(List<String> data) {
            System.out.println("数据处理业务逻辑");
        }
    }

}
